package com.czl.dxpro.dx_06_taskSix;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.czl.dxpro.dx_01_clear.TProperties;
public class dataMatchMapper extends Mapper<LongWritable,Text,NullWritable,Text> {
	private Map<String,TreeMap<String,String>> joinData = new HashMap<String,TreeMap<String,String>>();
	@Override
	protected void setup(Mapper<LongWritable, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
			//调用第一个缓存文件的文件路径
			Path file = new Path(context.getCacheFiles()[0].getPath());
			//读取file中的数据
			BufferedReader reader = new BufferedReader(new FileReader(file.toString()));
		String str = null;
		try {
			while((str = reader.readLine())!= null) {
				String[] splits = str.split(TProperties.getValue("fileoutsplit"));
			
			TreeMap<String,String> treemap = new TreeMap<String,String>(
					new StringComparator());
			if(joinData.containsKey(splits[1])) {
				joinData.get(splits[1]).put(splits[2]+","+splits[3]+","+splits[0]
					+","+splits[4]+","+splits[5],"");
			
			}
			else {
				treemap.put(splits[2]+","+splits[3]+","+splits[0]
						+","+splits[4]+","+splits[5],"");
				joinData.put(splits[1],treemap);
			}
			}
		}
		catch(Exception e){
			e.printStackTrace();
		}
		finally {
			reader.close();
		}
	}

	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		//从value中取得一级域名
		String[] values = value.toString().split(TProperties.getValue("fileoutsplit"));
		//根据域名从joinData中取得数据
		if(joinData.containsKey(values[1])) {
			TreeMap<String,String> treemap = joinData.get(values[1]);
			Iterator iterator = treemap.keySet().iterator();
			//按级别循环匹配
			while(iterator.hasNext()) {
				String[] krule = ((String) iterator.next()).split(",");
				if(values[2].indexOf(krule[0]) >=0) {
					StringBuffer sb = new StringBuffer();
					sb.append(krule[2]).append(TProperties.getValue("outfilesplit")).append(values[0])
					.append(TProperties.getValue("outfilesplit")).append(krule[3])
					.append(TProperties.getValue("outfilesplit")).append(values[2])
					.append(TProperties.getValue("outfilesplit")).append(krule[4]);
					
					
					Text ovalue = new Text(sb.toString());
					context.write(NullWritable.get(),ovalue);
					return;
				}
			}
		}
		
	}

	
}
